package org.funtester.performance.books.chapter09.section2;

import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.record.CompressionType;
import org.apache.kafka.common.serialization.StringSerializer;
import org.funtester.performance.books.chapter03.common.ThreadTool;

import java.util.Properties;

public class Produce {

    public static void main(String[] args) {
        Properties properties = new Properties();// 创建配置对象
        properties.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");// 指定kafka集群地址
        properties.setProperty(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());// 指定key序列化器
        properties.setProperty(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());// 指定value序列化器
        properties.setProperty(ProducerConfig.RETRIES_CONFIG, "3");// 重试次数
        properties.setProperty(ProducerConfig.ACKS_CONFIG, "all");// 0-不需要确认,1-需要leader确认,all-需要leader和follower确认
        properties.setProperty(ProducerConfig.BATCH_SIZE_CONFIG, "128");// 批次大小
        properties.setProperty(ProducerConfig.BUFFER_MEMORY_CONFIG, "33554432");// 缓冲区大小
        properties.setProperty(ProducerConfig.LINGER_MS_CONFIG, "100");// 等待时间,单位毫秒,默认是0,表示立即发送,设置成10表示10毫秒后发送
        properties.setProperty(ProducerConfig.COMPRESSION_TYPE_CONFIG, CompressionType.LZ4.name);// 压缩类型
        KafkaProducer<String, String> producer = new KafkaProducer<>(properties);// 创建生产者对象
        String topic = "FunTester";// 指定topic
        String key = "name";// 指定key
        String value = "小八超市会员";// 指定value
        ProducerRecord<String, String> record = new ProducerRecord<>(topic, key, value);// 创建消息对象

        producer.send(record);// 发送消息,同步方式
        for (int i = 0; i < 1000; i++) {
            producer.send(record, new Callback() {// 发送消息,异步回调
                @Override
                public void onCompletion(RecordMetadata recordMetadata, Exception e) {
                    if (e == null) {
                        System.out.println("消息发送成功！" + recordMetadata.offset());// 打印消息偏移量
                    } else {
                        System.err.println("消息发送失败！");
                    }
                }
            });
            ThreadTool.sleep(1000);
        }
        producer.close();// 关闭生产者对象
    }

}
